package com.wuwangfu.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashMap;

/**
 * @Author jcshen
 * @Date 2023-03-01
 * @PackageName:com.wuwangfu.state
 * @ClassName: CustomKeyedState
 * @Description: 自定义实现状态
 * @Version 1.0.0
 */
public class CustomKeyedState {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000 ));

        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);
        SingleOutputStreamOperator<Tuple2<String, Integer>> maped = line.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                if ("error".equals(value)) {
                    throw new RuntimeException("报错啦");
                }
                return Tuple2.of(value, 1);
            }
        });
        KeyedStream<Tuple2<String, Integer>, String> keyed = maped.keyBy(t -> t.f0);

        keyed.map(new RichMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
            //存储中间结果的一个集合
            private HashMap<String,Integer> counter = new HashMap<>();

            @Override
            public Tuple2<String, Integer> map(Tuple2<String, Integer> input) throws Exception {
                String word = input.f0;
                Integer count = input.f1;
                //从map中取出历史次数
                Integer histCount = counter.get(word);
                if (histCount == null){
                    histCount = 0;
                }
                //当前输入跟历史次数进行累加
                int sum = histCount + count;
                //更新map中的数据
                counter.put(word,sum);
                //输出结果
                return Tuple2.of(word,sum);
            }
        }).print();

        env.execute();
    }

}
